#include "MprpcProvider.h" // 这里直接写头文件，没写路径，就是得益于顶层CMake设置了头文件路径
#include "MprpcApplication.h"
#include "MprpcHeader.pb.h"
#include "Logger.h"

#include <functional>
#include <cstring>

// 框架提供给外部使用的，用来发布rpc方法的函数接口
void MprpcProvider::NotifyService(google::protobuf::Service *service)
{
    struct ServiceInfo service_info;
    service_info.m_service = service;

    // 获取 描述服务对象的指针，存储服务对象（UserServiceRpc）的一些属性
    const google::protobuf::ServiceDescriptor *pserviceDesc = service->GetDescriptor();
    // 从描述指针中 获取服务对象的名字 （FriendServiceRpc）
    std::string service_name = pserviceDesc->name();
    // 获取该服务对象所包含的 函数方法的个数
    int methodCnt = pserviceDesc->method_count();

    // std::cout << "service_name:" << service_name << std::endl;
    LOG_INFO_("service_name:%s", service_name.c_str());

    for (int i = 0; i < methodCnt; i++)
    {
        // 从服务对象描述中提取 函数方法描述指针
        const google::protobuf::MethodDescriptor *pmethodDesc = pserviceDesc->method(i);
        // 从服务函数方法描述中提取 函数方法名 （Login）
        std::string method_name = pmethodDesc->name();
        service_info.m_methodMap.insert(std::make_pair(method_name, pmethodDesc));

        // std::cout << "method_name:" << method_name << std::endl;
        LOG_INFO_("method_name:%s", method_name.c_str());
    }
    m_serviceMap.insert(std::make_pair(service_name, service_info));
}

// 启动rpc服务节点，开始提供rpc远程网络调用服务
void MprpcProvider::Run()
{
    std::string ip = MprpcApplication::GetInstance().GetConfig().Load("rpcServerIp");
    uint16_t port = atoi(MprpcApplication::GetInstance().GetConfig().Load("rpcServerPort").c_str());
    // muduo::net::InetAddress address(ip, port);
    InetAddress address(ip, port);

    // muduo的使用
    // 1.创建TcpServer对象
    // muduo::net::TcpServer server(&m_eventLoop, address, "MprpcProvider");
    TcpServer server(&m_eventLoop, address, "MprpcProvider");
    // 2.绑定回调函数
    server.setConnectionCallback(std::bind(&MprpcProvider::OnConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&MprpcProvider::OnMessage, this, std::placeholders::_1,
                                        std::placeholders::_2,
                                        std::placeholders::_3));

    // 3.设置线程数量
    server.setThreadNum(4);

    // 把当前rpc节点要发布的服务注册到zk上，让rpc client可以从zk上发现服务
    ZkClient zkclient;
    zkclient.Satrt();
    // service_name节点为永久性节点，method_name为临时性节点
    for (auto &sp : m_serviceMap)
    {
        // 先创建对象节点，再创建方法节点。  不能一起创建
        std::string service_path = "/" + sp.first; // 拼接对象节点路径
        zkclient.Create(service_path.c_str(), nullptr, 0);
        for (auto &mp : sp.second.m_methodMap)
        {
            std::string method_path = service_path + "/" + mp.first; // 拼接方法节点路径
            char znode_data[128] = {0};
            sprintf(znode_data, "%s:%d", ip.c_str(), port);
            // 这里注意，用的是strlen，只存有效的。
            zkclient.Create(method_path.c_str(), znode_data, strlen(znode_data), ZOO_EPHEMERAL);
        }
    }

    // std::cout << "MprpcProvider start service at ip:" << ip << " port:" << port << std::endl;
    LOG_INFO_("MprpcProvider start service at ip:%s port:%d", ip.c_str(), port);
    // 4.启动网络服务
    server.start();
    m_eventLoop.loop(); // 循环执行loop
}

// void MprpcProvider::OnConnection(const muduo::net::TcpConnectionPtr &conn)
void MprpcProvider::OnConnection(const TcpConnectionPtr &conn)
{
    if (!conn->connected())
    {
        conn->shutdown();
    }
}

/**
 * RpcProvider和RpcConsumer提前协商好 通信用的protobuf的数据格式
 * 前4个字节表示数据头（service_name + method_name）的长度
 * header_size + service_name method_name args_size + args  示例：16UserServiceLogin15zhang san123456   TODO
 * 在MprpcHeader.proto中定义好message类型，进行数据的序列化和反序列化
 */
// void MprpcProvider::OnMessage(const muduo::net::TcpConnectionPtr &conn,
//                               muduo::net::Buffer *buffer,
//                               muduo::Timestamp time)
void MprpcProvider::OnMessage(const TcpConnectionPtr &conn,
                            Buffer *buffer,
                            Timestamp time)
{
    // 从网络中接收的远程rpc调用请求序列化后的字符流
    std::string recv_buf = buffer->retrieveAllAsString();
    // 1.提取数据头的长度
    uint32_t header_size = 0;
    recv_buf.copy((char *)&header_size, 4, 0); // TO PONDER

    // 2.获得数据头的原始字符流（还没反序列化的）
    std::string rpc_header_str = recv_buf.substr(4, header_size);
    mprpc::MprpcHeader rpcHeader;

    std::string service_name;
    std::string method_name;
    uint32_t args_size = 0;
    if (rpcHeader.ParseFromString(rpc_header_str)) // 数据头原始字符流的反序列化
    {
        // 数据头反序列化成功
        // 3.获得服务对象名、函数方法名、参数长度
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else
    {
        // 数据头反序列化失败
        // std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
        // std::cout << strerror(errno) << std::endl;
        LOG_ERROR_("rpc_header_str parse error, rpc_header_str:%s. errno:%d :%s",
                   rpc_header_str.c_str(), errno, strerror(errno));
        return;
    }

    // 4.获得获取rpc方法参数的原始字符流
    std::string args_str = recv_buf.substr(4 + header_size, args_size);

    // 打印调试信息
    // std::cout << "======================================================" << std::endl;
    // std::cout << "原始字符流：" << recv_buf << std::endl;
    // std::cout << "header_size: " << header_size << std::endl;
    // std::cout << "rpc_header_str: " << rpc_header_str << std::endl;
    // std::cout << "service_name: " << service_name << std::endl;
    // std::cout << "method_name: " << method_name << std::endl;
    // std::cout << "args_size: " << args_size << std::endl;
    // std::cout << "args_str: " << args_str << std::endl;
    // std::cout << "======================================================" << std::endl;

    // 判断是否存在service_name对象
    std::unordered_map<std::string, struct ServiceInfo>::iterator service_it = m_serviceMap.find(service_name);
    if (service_it == m_serviceMap.end())
    {
        // std::cout << service_name << " service object does not exist!" << std::endl;
        LOG_ERROR_("%s service object does not exist!", service_name.c_str());
        return;
    }

    // 判断对象是否存在method_name方法
    auto method_it = service_it->second.m_methodMap.find(method_name);
    if (method_it == service_it->second.m_methodMap.end())
    {
        // std::cout << method_name << " method does not exist in the " << service_name << " service object" << std::endl;
        LOG_ERROR_("%s method does not exist in the %s", method_name.c_str(), service_name.c_str());
        return;
    }

    // 提取服务对象       即 服务提供方通过provider发布的new UserService
    google::protobuf::Service *service = service_it->second.m_service;
    const google::protobuf::MethodDescriptor *methodDesc = method_it->second; // 提取方法描述指针

    // 生成rpc方法调用的请求
    google::protobuf::Message *request = service->GetRequestPrototype(methodDesc).New();
    if (!request->ParseFromString(args_str)) // 反序列化成 响应对象
    {
        // std::cout << "request parse error!" << std::endl;
        LOG_ERROR_("request parse error!");
    }

    // 生成方法执行结果的响应
    google::protobuf::Message *response = service->GetResponsePrototype(methodDesc).New();

    // 函数模板的显示调用，<指明类型>（参数）
    // NewCallback返回一个Closure类对象。 作为下边CallMethod的最后一个参数
    // google::protobuf::Closure *done =
    //     google::protobuf::NewCallback<MprpcProvider, const muduo::net::TcpConnectionPtr &, google::protobuf::Message *>
    //                                 (this, &MprpcProvider::senRpcResponse, conn, response);

    google::protobuf::Closure *done =
        google::protobuf::NewCallback<MprpcProvider, const TcpConnectionPtr &, google::protobuf::Message *>
                                    (this, &MprpcProvider::senRpcResponse, conn, response);
    // 在框架上根据远端rpc请求，调用当前rpc节点上发布的方法。
    // Service类中的CallMethod是个纯虚函数，子类UserServiceRpc重写了，多态。
    // 执行UserServiceRpc::CallMethod，在其函数体内执行Login，而UserServiceRpc的Login是虚函数
    // 而其子类UserService类重写了Login，多态，成功调用方法并得到响应。
    /**
     * void Login(::google::protobuf::RpcController* controller,
                       const ::fixbug::LoginRequest* request,
                       ::fixbug::LoginResponse* response,
                       ::google::protobuf::Closure* done)
     */
    service->CallMethod(methodDesc, nullptr, request, response, done);
}

// void MprpcProvider::senRpcResponse(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
void MprpcProvider::senRpcResponse(const TcpConnectionPtr &conn, google::protobuf::Message *response)
{
    std::string response_str;                              // 存放响应序列化后的字符串
    if (response->SerializePartialToString(&response_str)) // 响应序列化
    {
        // conn->send(response_str.c_str(), response_str.size()); // muduo库网络发送
        conn->send(response_str.c_str(), response_str.size()); // muduo库网络发送
    }
    else
    {
        // std::cout << "Serialize error!" << std::endl;
        LOG_ERROR_("Serialize error!");
    }
    conn->shutdown(); // 成功失败均关闭连接
}
